package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.ChildScoreMode;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.BaseCategoryViewFeignClient;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserClientFeign;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@Slf4j
@Service
public class SearchServiceImpl implements SearchService {

    private final BaseCategoryViewFeignClient baseCategoryViewFeignClient;
    private final ElasticsearchClient elasticsearchClient;
    private final AlbumInfoFeignClient albumInfoFeignClient;
    private final UserClientFeign userClientFeign;
    private final ThreadPoolExecutor threadPoolExecutor;
    private final RBloomFilter<Object> rBloomFilter;

    @Autowired
    public SearchServiceImpl(BaseCategoryViewFeignClient baseCategoryViewFeignClient, ElasticsearchClient elasticsearchClient, AlbumInfoFeignClient albumInfoFeignClient, UserClientFeign userClientFeign, ThreadPoolExecutor threadPoolExecutor, RBloomFilter<Object> rBloomFilter) {
        this.baseCategoryViewFeignClient = baseCategoryViewFeignClient;
        this.elasticsearchClient = elasticsearchClient;
        this.albumInfoFeignClient = albumInfoFeignClient;
        this.userClientFeign = userClientFeign;
        this.threadPoolExecutor = threadPoolExecutor;
        this.rBloomFilter = rBloomFilter;
    }

    private static SearchRequest.Builder getBuilder(List<FieldValue> fieldValueList) {
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index("albuminfo_java0315");
        // 构建查询条件
        builder.query(query -> query.terms(
            terms -> terms.field("category3Id").terms(
                t -> t.value(fieldValueList)
            )
        ));
        // group by 条件：select id as aggId from a where a in (?,?,?,?,?,?,?) group by id
        builder.aggregations(
            // 第一次分桶聚合的别名
            "aggCategory3",
            agg -> agg.terms(
                    // 根据三级分类id进行分桶
                    terms -> terms.field("category3Id")
                )
                // 子聚合
                .aggregations(
                    "topHitsResult",
                    subAgg -> subAgg.topHits(
                        // 只获取前六条数据
                        t -> t.size(6).sort(
                            s -> s.field(
                                // 根据热度值排序，降序
                                f -> f.field("hotScore").order(SortOrder.Desc)))
                    )
                )
        );
        return builder;
    }

    /**
     * 首页频道数据：
     * 1. 查询到首页热门指定的七个三级分类
     * 2. 根据这七个三级分类的id去es中查询出这7个分类的全部专辑数据
     * 3. 根据三级分类id进行分桶（group by），然后在桶里面根据热门热度值进行降序取前6的数据
     *
     * @param category1Id
     * @return
     */
    @SneakyThrows
    @Override
    public Object channel(Long category1Id) {
        // 查询到首页热门指定的七个三级分类
        List<BaseCategory3> category3List = baseCategoryViewFeignClient.findTopBaseCategory3(category1Id);
        // 将这个三级分类的列表转换为一个map
        Map<Long, BaseCategory3> category3Map = category3List.stream().collect(Collectors.toMap(key -> key.getId(), value -> value));
        // 将三级分类列表转换为词条查询的查询条件类型
        List<FieldValue> fieldValueList = category3List.stream().map(category3 -> {
            return FieldValue.of(category3.getId());
        }).toList();
        // 搜索请求对象初始化
        SearchRequest.Builder builder = getBuilder(fieldValueList);
        // 搜索
        SearchResponse<AlbumInfoIndex> searchResponse =
            elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);
        // 解析结果
        Map<String, Aggregate> aggregations = searchResponse.aggregations();
        if (aggregations != null && !aggregations.isEmpty()) {
            // 获取分桶的结果
            return aggregations.get("aggCategory3").lterms().buckets().array().stream().map(longTermsBucket -> {
                JSONObject c3Json = new JSONObject();
                // 获取三级分类id
                long category3Id = longTermsBucket.key();
                c3Json.put("baseCategory3", category3Map.get(category3Id));
                // 获取桶里面的数据（子分类的数据）
                List<AlbumInfoIndex> albumInfoIndexList = longTermsBucket.aggregations().get("topHitsResult").topHits().hits().hits().stream().map(hit -> {
                    if (hit.source() != null) {
                        return hit.source().to(AlbumInfoIndex.class);
                    }
                    return null;
                }).toList();
                c3Json.put("list", albumInfoIndexList);
                // 返回
                return c3Json;
            }).toList();
        }
        // 返回
        return null;
    }

    /**
     * 专辑搜索
     *
     * @param albumIndexQuery
     * @return
     */
    @Override
    @SneakyThrows
    public Object search(AlbumIndexQuery albumIndexQuery) {
        // 拼接条件
        SearchRequest searchRequest = buildQueryParams(albumIndexQuery);
        // 执行搜索
        SearchResponse<AlbumInfoIndex> searchResponse =
            elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
        // 解析结果
        AlbumSearchResponseVo result = getSearchResult(searchResponse);
        result.setPageNo(albumIndexQuery.getPageNo());
        Integer pageSize = albumIndexQuery.getPageSize();
        result.setPageSize(pageSize);
        // 计算总页数
        Long total = result.getTotal();
        Long pages = total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        result.setTotalPages(pages);
        // 返回结果
        return result;
    }

    /**
     * 拼接条件
     *
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest buildQueryParams(AlbumIndexQuery albumIndexQuery) {
        // 请求对象初始化
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 指定索引
        builder.index("albuminfo_java0315");
        // 构建组合查询对象
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        // 关键字
        String keyword = albumIndexQuery.getKeyword();
        if (StringUtils.isNotEmpty(keyword)) {
            boolQueryBuilder
                // 专辑的名字
                .should(s -> s.match(m -> m.field("albumTitle").query(keyword)))
                // 专辑的描述
                .should(s -> s.match(m -> m.field("albumIntro").query(keyword)));
        }
        // 一级分类
        Long category1Id = albumIndexQuery.getCategory1Id();
        if (category1Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        // 二级分类
        Long category2Id = albumIndexQuery.getCategory2Id();
        if (category2Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));
        }
        // 三级分类
        Long category3Id = albumIndexQuery.getCategory3Id();
        if (category3Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
        }
        // 标签条件：多个之间是must关系
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (attributeList != null && !attributeList.isEmpty()) {
            attributeList.stream().forEach(attribute -> {
                // 切分：0-属性id must 1-属性值的id
                String[] split = attribute.split(":");
                // 添加条件
                boolQueryBuilder.filter(
                    f -> f.nested(
                        nested -> nested.query(
                            q -> q.bool(
                                b -> b
                                    .must(m -> m.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
                                    .must(m -> m.term(t -> t.field("attributeValueIndexList.valueId").value(split[1])))
                            )
                        ).path("attributeValueIndexList").scoreMode(ChildScoreMode.None)
                    )
                );
            });
        }

        // 存储查询条件
        builder.query(boolQueryBuilder.build()._toQuery());
        // 分页
        Integer pageNo = albumIndexQuery.getPageNo();
        Integer pageSize = albumIndexQuery.getPageSize();
        builder.from((pageNo - 1) * pageSize);
        builder.size(pageSize);
        // 排序
        String order = albumIndexQuery.getOrder();
        if (StringUtils.isNotEmpty(order)) {
            // 切分：0-域 1-规则
            String[] split = order.split(":");
            switch (split[0]) {
                case "1" ->
                    builder.sort(s -> s.field(f -> f.field("id").order("desc".equals(split[1]) ? SortOrder.Desc : SortOrder.Asc)));
                case "2" ->
                    builder.sort(s -> s.field(f -> f.field("playStatNum").order("desc".equals(split[1]) ? SortOrder.Desc : SortOrder.Asc)));
                case "3" ->
                    builder.sort(s -> s.field(f -> f.field("createTime").order("desc".equals(split[1]) ? SortOrder.Desc : SortOrder.Asc)));
                default -> builder.sort(s -> s.field(f -> f.field("id").order(SortOrder.Desc)));
            }
        }
        // 高亮
        builder.highlight(
            h -> h.fields("albumTitle", f -> f.preTags("<font style=color:red>").postTags("</font>"))
        );
        // 返回
        return builder.build();
    }

    /**
     * 解析结果
     *
     * @param searchResponse
     * @return
     */
    private AlbumSearchResponseVo getSearchResult(SearchResponse<AlbumInfoIndex> searchResponse) {
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        // 获取命中的数据对象
        HitsMetadata<AlbumInfoIndex> hits = searchResponse.hits();
        // 获取总命中的数据
        long value = hits.total().value();
        albumSearchResponseVo.setTotal(value);
        // 获取专辑数据
        List<AlbumInfoIndexVo> albumInfoIndexList = hits.hits().stream().map(
            h -> {
                // 专辑的原始数据
                AlbumInfoIndex albumInfoIndex = h.source();
                // 获取高亮的数据
                Map<String, List<String>> highlight = h.highlight();
                if (highlight != null && !highlight.isEmpty()) {
                    List<String> albumTitleList = highlight.get("albumTitle");
                    String albumTitle = "";
                    for (String s : albumTitleList) {
                        albumTitle += s;
                    }
                    albumInfoIndex.setAlbumTitle(albumTitle);
                }
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                return albumInfoIndexVo;
            }
        ).toList();
        albumSearchResponseVo.setList(albumInfoIndexList);
        // 包装返回结果
        return albumSearchResponseVo;
    }

    /**
     * 提示词接口
     *
     * @param keywords
     * @return
     */
    @SneakyThrows
    @Override
    public Object completeSuggest(String keywords) {
        // 判断
        if (StringUtils.isEmpty(keywords)) {
            return null;
        }
        // 请求对象
        SearchRequest.Builder builder = new SearchRequest.Builder();
        // 设置索引
        builder.index("suggestinfo_java0315");
        // 构建条件
        builder.suggest(
            suggest -> suggest
                .suggesters(
                    "suggestKeyword",
                    f -> f
                        // 前置匹配
                        .prefix(keywords)
                        .completion(
                            c -> c
                                // 需要匹配的域（字段）
                                .field("keyword")
                                // 去重
                                .skipDuplicates(true)
                                // 允许出现偏差
                                .fuzzy(
                                    fu -> fu.fuzziness("auto")
                                )
                        )
                )
                .suggesters(
                    "suggestKeywordPinYin",
                    f -> f
                        // 前置匹配
                        .prefix(keywords)
                        .completion(
                            c -> c
                                // 需要匹配的域（字段）
                                .field("keywordPinyin")
                                // 去重
                                .skipDuplicates(true)
                                // 允许出现偏差
                                .fuzzy(
                                    fu -> fu.fuzziness("auto")
                                )
                        )
                )
                .suggesters(
                    "suggestKeywordSequence",
                    f -> f
                        // 前置匹配
                        .prefix(keywords)
                        .completion(
                            c -> c
                                // 需要匹配的域（字段）
                                .field("keywordSequence")
                                // 去重
                                .skipDuplicates(true)
                                // 允许出现偏差
                                .fuzzy(
                                    fu -> fu.fuzziness("auto")
                                )
                        )
                )

        );
        SearchRequest searchRequest = builder.build();
        // 搜索提示词
        SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(searchRequest, SuggestIndex.class);
        // 解析结果
        Map<String, List<Suggestion<SuggestIndex>>> suggestResult = searchResponse.suggest();
        if (suggestResult != null && !suggestResult.isEmpty()) {
            // 解析关键字的匹配结果
            List<String> list1 = suggestResult.get("suggestKeyword").get(0).completion().options().stream().map(
                suggestOption -> {
                    SuggestIndex source = suggestOption.source();
                    return source.getTitle();
                }
            ).toList();
            // 解析关键字拼音的匹配结果
            List<String> list2 = suggestResult.get("suggestKeywordPinYin").get(0).completion().options().stream().map(
                suggestOption -> {
                    SuggestIndex source = suggestOption.source();
                    return source.getTitle();
                }
            ).toList();
            // 解析关键字首字母的匹配结果
            List<String> list3 = suggestResult.get("suggestKeywordSequence").get(0).completion().options().stream().map(
                suggestOption -> {
                    SuggestIndex source = suggestOption.source();
                    return source.getTitle();
                }
            ).toList();
            // 三个列表和为一个，去重，限制10条
            return Stream.of(list1, list2, list3)
                // 合并
                .flatMap(Collection::stream)
                // 去重
                .collect(Collectors.toSet())
                // 限制10条
                .stream().limit(10)
                .toList();
        }
        return null;
    }

    /**
     * 查询专辑的详细信息
     *
     * @param albumId
     * @return
     */
    @Override
    public Object albumDetail(Long albumId) {
        Map<String, Object> result = new ConcurrentHashMap<>();
        // 前往布隆过滤器确认
        // if (!rBloomFilter.contains(albumId)) {
        //     return result;
        // }
        // 查询专辑的数据
        CompletableFuture<AlbumInfo> future1 = CompletableFuture.supplyAsync(() -> {
            AlbumInfo albumInfo = albumInfoFeignClient.getAlbumInfoById(albumId);
            if (albumInfo == null || albumInfo.getId() == null) {
                return null;
            }
            result.put("albumInfo", albumInfo);
            return albumInfo;
        }, threadPoolExecutor);

        // 专辑的统计信息
        CompletableFuture<Void> future2 = future1.thenAcceptAsync(albumInfo -> {
            if (albumInfo == null || albumInfo.getId() == null) {
                return;
            }
            Map<String, Object> albumStat = albumInfoFeignClient.getAlbumStat(albumId);
            AlbumStatVo albumStatVo = new AlbumStatVo();
            albumStatVo.setAlbumId(albumId);
            albumStatVo.setPlayStatNum(Integer.valueOf(albumStat.get(SystemConstant.ALBUM_STAT_PLAY).toString()));
            albumStatVo.setSubscribeStatNum(Integer.valueOf(albumStat.get(SystemConstant.ALBUM_STAT_SUBSCRIBE).toString()));
            albumStatVo.setBuyStatNum(Integer.valueOf(albumStat.get(SystemConstant.ALBUM_STAT_BROWSE).toString()));
            albumStatVo.setCommentStatNum(Integer.valueOf(albumStat.get(SystemConstant.ALBUM_STAT_COMMENT).toString()));
            result.put("albumStatVo", albumStatVo);
        }, threadPoolExecutor);

        // 专辑的作者信息
        CompletableFuture<Void> future3 = future1.thenAcceptAsync(albumInfo -> {
            if (albumInfo == null || albumInfo.getId() == null) {
                return;
            }
            UserInfo userInfo = userClientFeign.getUserInfo(albumInfo.getUserId());
            UserInfoVo userInfoVo = new UserInfoVo();
            BeanUtils.copyProperties(userInfo, userInfoVo);
            result.put("announcer", userInfoVo);
        }, threadPoolExecutor);

        // 专辑的分类信息
        CompletableFuture<Void> future4 = future1.thenAcceptAsync(albumInfo -> {
            if (albumInfo == null || albumInfo.getId() == null) {
                return;
            }
            BaseCategoryView baseCategoryView = baseCategoryViewFeignClient.getBseCategoryView(albumInfo.getCategory3Id());
            result.put("baseCategoryView", baseCategoryView);
        }, threadPoolExecutor);
        // 所有任务结束
        CompletableFuture.allOf(future2, future3, future4).join();
        // 返回
        return result;
    }
}
